package com.weiyu.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * @author weiyu@gomeholdings.com
 * SWSSS@description
 * @create 2017/6/29
 */
public class SimpleProducer implements Runnable{
    private static final Logger logger = LoggerFactory.getLogger(SimpleProducer.class);
    private static final String TOPIC = "test";
    private Producer<String,String> producer;

    public SimpleProducer(){
        Properties props = new Properties();
        props.put("bootstrap.servers","XXX");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        //key的序列化类
        props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        //value的序列化类
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer<>(props);
    }

    public static void main(String[] args) {
        Thread t1 = new Thread(new SimpleProducer());
        t1.start();
    }

    @Override
    public void run() {
        int i = 0;
        while (true){
            String str = "this is a test"+i++;
            System.out.println("生产信息："+str);
            producer.send(new ProducerRecord(TOPIC,"test111",str));
            try {
                Thread.sleep(1000);
            }
            catch (InterruptedException e) {
                logger.error(e.getLocalizedMessage());
                producer.close();
            }
        }
    }
}
